package com.cxm.weilai.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.io.IOException;
import java.util.Properties;

/**
 * kafka生产者
 *
 * @ProjectName: [kafka-apache-java]
 * @Package: [cn.lichn.kafka.producer]
 * @ClassName: [Producer]
 * @Description: [一句话描述该类的功能]
 * @Author: [chenman]
 * @CreateDate: [2017/11/8 14:38]
 * @UpdateUser: [chenman]
 * @UpdateDate: [2017/11/8 14:38]
 * @UpdateRemark: [说明本次修改内容]
 * @Version: [v1.0]
 */
public class Producer1 {

    private static KafkaProducer kafkaProducer = null;
    private static Producer1 producer = null;

    public static Producer1 getInstance() {
        if (producer == null) {
            synchronized (Producer1.class) {
                if (producer == null) {
                    producer = new Producer1();
                }
            }
        }
        return producer;
    }

    /**
     * kafka producer初始化操作
     *
     * @param configFile
     * @throws IOException
     */
    public void init(String configFile) throws IOException {
        Properties properties = new Properties();
        properties.load(this.getClass().getClassLoader().getResourceAsStream(configFile));
        kafkaProducer = new KafkaProducer(properties);
    }

    /**
     * 发送一条消息
     *
     * @param topic
     * @param message
     */
    public void send(String topic, String message) {
        ProducerRecord record;
        record = new ProducerRecord(topic, message, message);
        kafkaProducer.send(record);
        // kafkaProducer.flush();
    }

}